package com.hba.producer;

import com.hba.broker.RabbitBroker;
import com.hba.constant.RabbitConstant;
import com.hba.consumer.MessageBody;
import com.hba.util.HessionCodecFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.amqp.support.postprocessor.AbstractCompressingPostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.OutputStream;
import java.util.Random;
import java.util.UUID;

@Component
public class MessageProducer extends RabbitBroker {

    private static final Logger logger = LoggerFactory.getLogger(MessageProducer.class);

    @Autowired
    RabbitTemplate rabbitTemplate;

    public void sendMessage(String queueName,Object message) throws Exception{
        String exchangeName = RabbitConstant.DEFAULT_EXCHANGE_NAME;
        this.sendMessage(exchangeName,queueName,message);
    }

    public void sendMessage(String exchangeName,String queueName,Object message) throws Exception{
        declareBindExchange(exchangeName,queueName,RabbitConstant.DEFAULT_DEAD_LETTER_EXCHANGE,queueName+RabbitConstant.DEFAULT_SEQUEUE_RETRY_SUFFIX);
        byte[] messageByte = HessionCodecFactory.serialize(message);
        MessageBody messageBody=new MessageBody(queueName,RabbitConstant.DEFAULT_EXCHANGE_NAME,messageByte, UUID.randomUUID().toString());
        Random random = new Random();
        int index = random.nextInt(num);
        String logicQueue = queueName + "_" + index;
        rabbitTemplate.convertAndSend(exchangeName,logicQueue,messageBody);
    }

    public void sendMessageWithDelay(String queueName,Object message,final int millisecond) throws Exception{
        String exchangeName = RabbitConstant.DEFAULT_EXCHANGE_NAME;
        this.sendMessageWithDelay(exchangeName,queueName,message,millisecond);
    }

    public void sendMessageWithDelay(String exchangeName,String queueName,Object message,final int millisecond) throws Exception{
        declareBindExchange(RabbitConstant.DEFAULT_DELAY_EXCHANGE_NAME,queueName+RabbitConstant.DEFAULT_SEQUEUE_DELAY_SUFFIX,exchangeName,queueName);
        declareBindExchange(exchangeName,queueName,RabbitConstant.DEFAULT_DEAD_LETTER_EXCHANGE,queueName+RabbitConstant.DEFAULT_SEQUEUE_RETRY_SUFFIX);
        declareDeadLetterBindExchange(RabbitConstant.DEFAULT_DEAD_LETTER_EXCHANGE,queueName+RabbitConstant.DEFAULT_SEQUEUE_RETRY_SUFFIX,RabbitConstant.DEFAULT_DELAY_EXCHANGE_NAME,queueName+RabbitConstant.DEFAULT_SEQUEUE_DELAY_SUFFIX);
        byte[] messageByte = HessionCodecFactory.serialize(message);
        MessageBody messageBody=new MessageBody(queueName,RabbitConstant.DEFAULT_EXCHANGE_NAME,messageByte, UUID.randomUUID().toString());
        MessageConverter messageConverter = new SimpleMessageConverter();
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setExpiration(String.valueOf(millisecond));
        Object messageObject = messageConverter.toMessage(messageBody,messageProperties);
        rabbitTemplate.convertAndSend(RabbitConstant.DEFAULT_DELAY_EXCHANGE_NAME,queueName+RabbitConstant.DEFAULT_SEQUEUE_DELAY_SUFFIX,messageObject);
    }
}
